package njuflink2;

import njuflink.GlobalConfig;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class MySQLSink2_1 extends RichSinkFunction<Tuple2<String,Long>> {    //private Connection connection=njuflink.KafkaFlinkMySQL.connection;
    private PreparedStatement preparedStatement;
    private String tb= njuflink2.GlobalConfig.TABLE2_1;
    private Connection connection;

    @Override
    public void open (org.apache.flink.configuration.Configuration parameters) throws Exception{

//        //加载JDBC驱动
//        Class.forName(njuflink.GlobalConfig.DRIVER_CLASS);
        //获取数据库链接
        connection= DriverManager.getConnection(njuflink.GlobalConfig.DB_URL, njuflink.GlobalConfig.USER_NAME, GlobalConfig.PASSWORD);
        // 关闭自定提交
        connection.setAutoCommit(false);

        System.out.println("sink db connected");

    }
    @Override
    public void close() throws Exception{
        if(preparedStatement!=null){
            preparedStatement.close();
        }
        if(connection!=null){
            connection.close();
        }
    }

    public void invoke(Tuple2<String,Long> value, Context context)throws Exception{
        if(value==null){
            return;
        }
        System.out.println(value);


        String lang=value.f0;
        Long count=value.f1;

        String sort=" order by count desc";
        String querySQL="select * from "+tb+" where language = '"+lang+"';";
        String updateSQL="update "+tb+" set count = "+count+" where language = '"+lang+"' ";
        String insertSQL="insert into "+tb+"(language,count) values('"+lang+"', "+count+") ";

        //查询数据
        preparedStatement= connection.prepareStatement(querySQL);
        //System.out.println(querySQL);
        ResultSet resultSet=preparedStatement.executeQuery();
        if(resultSet.next()){
            //更新数据
            //System.out.println(updateSQL);
            preparedStatement.executeUpdate(updateSQL);
        }else{
            //插入数据
            //System.out.println(insertSQL);
            preparedStatement.execute(insertSQL);
        }
        // 提交
        connection.commit();


    }

}


